package com.soup.lab.basic.iso.executor;

import com.segi.uhomecp.park.access.util.empty.EmptyUtils;
import com.segi.uhomecp.park.access.util.lock.RedissonUtils;
import org.redisson.api.RLock;

import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.stream.Stream;

/**
 * 互斥键集放在Redis中，以实现多进程互斥键集共享
 *
 * @author soup
 */
@SuppressWarnings("rawtypes")
public class RedisKeyPickThreadPoolExecutor extends AbstractKeyPickThreadPoolExecutor {

    private final Map<String, RLock> lockMap = new ConcurrentHashMap<>();

    @Override
    protected void removeKeys(Collection<?> keys) {
        String[] validKeys = convertKeys(keys);
        if(EmptyUtils.isEmpty(validKeys)) {
            return;
        }
        Stream.of(validKeys).map(lockMap::remove).distinct().forEach(RLock::forceUnlockAsync);
    }

    @Override
    protected boolean addKeysIfNotExists(Collection<?> keys) {
        String[] validKeys = convertKeys(keys);
        if(EmptyUtils.isEmpty(validKeys)) {
            return true;
        }
        Map<String, RLock> lockMap = this.lockMap;
        if(Stream.of(validKeys).anyMatch(lockMap::containsKey)) {
            return false;
        }
        RLock lock = RedissonUtils.tryLockMulti(validKeys);
        if(lock == null) {
            return false;
        }
        Stream.of(validKeys).forEach(k -> lockMap.put(k, lock));
        return true;
    }

    protected String[] convertKeys(Collection<?> keys) {
        return keys.stream().filter(Objects::nonNull).map(Objects::toString).distinct().toArray(String[]::new);
    }

    @Override
    public void clearKeys() {
       lockMap.values().stream().distinct().forEach(RLock::forceUnlockAsync);
       lockMap.clear();
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                           RejectedExecutionHandler handler, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, scannerThreadName, scannerThreadAwakeTimeout);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                           RejectedExecutionHandler handler, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName, long scannerThreadAwakeTimeout) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, waitingQueue, scannerThreadName, scannerThreadAwakeTimeout);
    }

    private RedisKeyPickThreadPoolExecutor(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName) {
        super(corePoolSize, maximumPoolSize, workQueue, waitingQueue, scannerThreadName);
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                        RejectedExecutionHandler handler, String scannerThreadName, long scannerThreadAwakeTimeout) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, scannerThreadName, scannerThreadAwakeTimeout);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                                        RejectedExecutionHandler handler, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName, long scannerThreadAwakeTimeout) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler, waitingQueue, scannerThreadName, scannerThreadAwakeTimeout);
        executor.start();
        return executor;
    }

    public static RedisKeyPickThreadPoolExecutor create(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue, BlockingQueue<KeyPickRunnableFuture> waitingQueue, String scannerThreadName) {
        RedisKeyPickThreadPoolExecutor executor = new RedisKeyPickThreadPoolExecutor(corePoolSize, maximumPoolSize, workQueue, waitingQueue, scannerThreadName);
        executor.start();
        return executor;
    }

}
